package com.pxene.dmp.reducer;

import java.io.IOException;
import java.util.Date;
import java.util.HashSet;
import java.util.Set;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class GetSearchWordReducer extends Reducer<Text, Text, LongWritable, Text> {
	
	private LongWritable resultKey = new LongWritable();
	private Text resultVal = new Text();
	
	private Set<String> wordSet = new HashSet<String>();
	
	@Override
	protected void setup(Context context) throws IOException, InterruptedException {
		Configuration conf = context.getConfiguration();
		// 设置配置文件名
		System.setProperty("config-file", conf.get("config-file"));
	}

	@Override
	protected void reduce(Text key, Iterable<Text> values, Context context)
			throws IOException, InterruptedException {
		String[] columns = key.toString().split("\\|");
		if (columns.length != 5) {
			return;
		}
		String imsi = columns[0];
		String bsid = columns[1];
		String starttime = columns[2];
		String province = columns[3];
		String dataType = columns[4];
		
		wordSet.clear();
		for (Text value : values) {
			String word = value.toString();
			if (!"".equals(word)) {
				wordSet.add(word);
			}
		}
		
		// 防止记录过长，一条记录只包含3个word
		StringBuilder sb = new StringBuilder();
		int count = 0;
		for (String word : wordSet) {
			sb.append(word).append(";");
			count ++;
			if (count == 3) {
				resultKey.set(new Date().getTime());
				resultVal.set(imsi + "|" + bsid + "|" + starttime 
						+ "|" + sb.substring(0, sb.length()-1) + "|" + province + "|" + dataType);
				context.write(resultKey, resultVal);
				sb.delete(0, sb.length());
				count = 0;
			}
		}
		if (count > 0) {
			resultKey.set(new Date().getTime());
			resultVal.set(imsi + "|" + bsid + "|" + starttime 
					+ "|" + sb.substring(0, sb.length()-1) + "|" + province + "|" + dataType);
			context.write(resultKey, resultVal);
		}
		
	}

}
